package com.ibs.hadoop.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 进行hbase的连接操作信息
 * 
 * @author liujun
 *
 */
public class HbaseClient {

    /**
     * 行结尾标识，用于在进行分页时排除当前行
     */
    final byte[] POSTFIX = new byte[] { 0x00 };

    /**
     * zookeeper的地址
     */
    private static final String ZOOKEEPERURL = "www.os1.com,www.os2.com,www.os3.com";

    /**
     * zookeeper的连接端口
     */
    private static final String CLIENTPORT = "2181";

    /**
     * 远程连接配制信息
     */
    private final Configuration conf;

    /**
     * 连接信息
     */
    private Connection hbaseConn;

    public HbaseClient() {
        // 得到配制信息
        conf = HBaseConfiguration.create();
        // 设置zookeeper的连接信息
        conf.set("hbase.zookeeper.quorum", ZOOKEEPERURL);
        conf.set("hbase.zookeeper.property.clientPort", CLIENTPORT);

        // 得到连接信息
        try {
            hbaseConn = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 进行查询操作
     * 
     * @param table
     * @return
     */
    public List<Map<String, Object>> query(String tableName, String rowKey) {
        HTable table = null;
        Result rs = null;
        try {
            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));
            Get key = new Get(rowKey.getBytes());
            // 测试仅提取十行记录
            key.setAttribute("LIMIT", "10".getBytes());
            rs = table.get(key);

            for (Cell cell : rs.listCells()) {
                System.out.println("列族名:" + new String(CellUtil.cloneQualifier(cell)));
                System.out.println("列族值:" + new String(CellUtil.cloneValue(cell)));
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 进行分页的总行数统计
     * 
     * @return
     */
    public long queryCount(String tableName, String startRow, String stopRow) {

        AggregationClient aggregationClient = new AggregationClient(conf);
        try {

            Scan scan = new Scan();

            scan.setStartRow(startRow.getBytes());
            scan.setStopRow(stopRow.getBytes());

            long rowCount = aggregationClient.rowCount(TableName.valueOf(tableName), new LongColumnInterpreter(), scan);

            return rowCount;
        } catch (IOException e) {
            e.printStackTrace();
        } catch (Throwable e) {
            e.printStackTrace();
        } finally {
            try {
                aggregationClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        return 0;
    }

    /**
     * 进行查询操作
     * 
     * @param table
     * @return
     */
    public List<Map<String, Object>> queryScaner(String tableName, int rowNum) {
        HTable table = null;
        ResultScanner rs = null;
        try {
            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Scan scan = new Scan();

            // 过滤掉的行数
            scan.setFilter(new PageFilter(rowNum));

            scan.setMaxResultSize(rowNum);

            rs = table.getScanner(scan);

            int index = 0;
            for (Result result : rs) {
                Cell[] kv = result.rawCells();
                System.out.print("当前第" + index + "行,");

                for (int i = 0; i < kv.length; i++) {
                    System.out.print("行键:" + new String(CellUtil.cloneRow(kv[i])) + ",");
                    System.out.print("列族的名称:" + new String(CellUtil.cloneFamily(kv[i])) + ",");
                    System.out.print("列下列的名称:" + new String(CellUtil.cloneQualifier(kv[i])) + ",");

                    System.out.print("列的值:" + new String(CellUtil.cloneValue(kv[i])));

                }
                System.out.println();
                index++;
            }

        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * 进行分页查询操作
     * 
     * @param tableName
     * @param rowNum
     * @param startRow
     * @param stopRow
     * @param endRowBytes
     * @return
     */
    public List<DnsQueryLteInfo> queryPage(String tableName, int rowNum, String startRow, String stopRow,
            byte[] endRowBytes) {
        // 行结束
        HTable table = null;
        ResultScanner rs = null;
        List<DnsQueryLteInfo> list = null;

        try {
            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Scan scan = new Scan();

            // 满足所有条件
            FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
            // 设置过滤的行数
            Filter filterPage = new PageFilter(rowNum);
            filterList.addFilter(filterPage);

            // 设置过滤信息
            scan.setFilter(filterList);
            scan.setMaxResultSize(rowNum);

            if (null != endRowBytes) {
                byte[] row = Bytes.add(endRowBytes, POSTFIX);
                scan.setStartRow(row);
            } else {
                scan.setStartRow(Bytes.toBytes(startRow));
            }

            if (StringUtils.isEmpty(stopRow)) {
                scan.setStopRow(Bytes.toBytes(stopRow));
            }

            rs = table.getScanner(scan);

            list = new ArrayList<>(rowNum);

            int index = 0;
            for (Result result : rs) {
                Cell[] kv = result.rawCells();
                System.out.print("当前第" + index + "行,");

                for (int i = 0; i < kv.length; i++) {
                    System.out.print("行键:" + new String(CellUtil.cloneRow(kv[i])) + ",");
                    System.out.print("列族的名称:" + new String(CellUtil.cloneFamily(kv[i])) + ",");
                    System.out.print("列下列的名称:" + new String(CellUtil.cloneQualifier(kv[i])) + ",");
                    System.out.print("列的值:" + new String(CellUtil.cloneValue(kv[i])));
                    list.add(new DnsQueryLteInfo(new String(CellUtil.cloneValue(kv[i]))));
                }
                System.out.println();
                index++;
                endRowBytes = result.getRow();

            }

            System.out.println("最后一条:" + new String(endRowBytes));

        } catch (IOException e) {
            e.printStackTrace();
        }

        return list;
    }

    public void testPutDate() {
        String tableName = "LTE_DNS_QUERY_20160829";

        HTable table = null;
        try {
            table = (HTable) hbaseConn.getTable(TableName.valueOf(tableName));

            Put put = new Put("460079146537649789456131231123".getBytes());

            put.addImmutable("cf1".getBytes(), "c".getBytes(),
                    "36838142,37371817,2016-05-08 09:14:42,1107334368,,,2886810325,2026153223,1682045413,1681964884,FOSSAEGW1001BEr,,182568450,,sdkctllbs.yy.com,,,Time out,4,2971729008,1,2016-05-08 09:14:52,,44049,,9540,,,,"
                            .getBytes());

            table.put(put);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        HbaseClient client = new HbaseClient();
        // client.queryScaner("LTE_DNS_QUERY_20160829", 100);

        // List<DnsQueryLteInfo> list = null;
        //
        // System.out.println("进行分页...");
        // byte[] endRow = null;
        // for (int i = 0; i < 10; i++) {
        // System.out.println("当前第:" + i + "页");
        // list = client.queryPage("LTE_DNS_QUERY_20160829", 10,
        // "00000357200006420160508080449000108",
        // "999999529850006420160508080753000638", endRow);
        // }
        //
        // list.forEach(s -> System.out.println(s));
        //
        // System.out.println("当前集合信息:" + list.size());

        client.query("LTE_GET_160829", "00000787620006420160");

        // client.testPutDate();
    }
}
